package com.quanyan.ic.datacenter.mq;

import com.alibaba.fastjson.JSON;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.quanyan.ic.datacenter.constant.MqConstants;
import com.quanyan.ic.datacenter.email.MailService;
import com.quanyan.ic.datacenter.model.SerialPortExp;
import com.quanyan.ic.middleware.mq.rocketmq.consumer.AbstractMqConsumer;
import com.quanyan.ic.middleware.mq.rocketmq.consumer.MqConsumerStatus;
import com.quanyan.ic.middleware.mq.rocketmq.consumer.MqTopic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;

/**
 * Created by zhuo.shi on 2017/8/23.
 */
@Component
public class LightSerialPortExpConsumer extends AbstractMqConsumer {

    private static Logger logger = LoggerFactory.getLogger(LightSerialPortExpConsumer.class);

    @Value("${rocketmq.nameServer}")
    private String nameServer;

    @Autowired
    private MailService mailService;

    @PostConstruct
    public void start(){
        super.start();
        logger.info("LightSerialPortExpConsumer started!");
    }

    @Override
    protected MqConsumerStatus process(List<MessageExt> messages, ConsumeConcurrentlyContext context) {
        logger.info("received SerialPortExp MQ msg: {}", messages);
        if (messages.isEmpty()){
            return MqConsumerStatus.CONSUMER_SUCCESS;
        }
        MessageExt messageExt = messages.get(0);
        byte[] bodyByte = messageExt.getBody();
        String str = new String(bodyByte);
        logger.info("SerialPortExp={}", str);
        SerialPortExp serialPortExp = JSON.parseObject(str, SerialPortExp.class);
        mailService.sendSimpleMail("monitorgroup.ic@quncaotech.com",
                "SerialPortExp | " + serialPortExp.getPlaceName() + "（"+serialPortExp.getPlaceId()+"）" + " | " +serialPortExp.getReason(),
                str);
        return MqConsumerStatus.CONSUMER_SUCCESS;
    }

    @Override
    protected String getConsumerGroup() {
        return MqConstants.GROUP_C_SERIALPORTEXP;
    }

    @Override
    protected String getNameServerAddress() {
        return nameServer;
    }

    @Override
    protected String getInstanceName() {
        return null;
    }

    @Override
    protected MqTopic[] getMqTopics() {
        MqTopic[] mqTopics = new MqTopic[1];
        mqTopics[0] = new MqTopic(
                MqConstants.TOPIC_IC, MqConstants.TAGS_LIGHT_SERIALPORTEXP);
        return mqTopics;
    }
}
